package com.eshop.eshopcache.kafka;

import com.eshop.eshopcache.service.CacheService;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author: Xukai
 * @description:
 * @createDate: 2018/7/26 10:24
 * @modified By:
 */
@Slf4j
@Component
public class KafkaConsumer {

    @Autowired
    private CacheService cacheService;

    /**
     * 监听eshop主题,有消息就读取
     */
    @KafkaListener(topics = {"eshop"})
    public void consumer(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            String message = kafkaMessage.get().toString();
            log.info("Get message from kafka: {}", message);
            ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("KafkaMessageProcessor").build();
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                    1,
                    1,
                    0L,
                    TimeUnit.MILLISECONDS,
                    new LinkedBlockingDeque<>(),
                    namedThreadFactory,
                    new ThreadPoolExecutor.AbortPolicy());
            threadPoolExecutor.submit(new KafkaMessageProcessor(message, cacheService));
        }
    }

}
